package iredis

import (
	"fmt"
	"gitee.com/tietang/terrace-go/v2/base"
	"github.com/go-redis/redis/v8"
	"github.com/go-redsync/redsync/v4"
	"github.com/go-redsync/redsync/v4/redis/goredis/v8"
	log "github.com/sirupsen/logrus"
	"github.com/tietang/go-utils"
	"github.com/tietang/props/v3/kvs"
	"time"
)

const (
	CLIENT_MODE_NEW = 0
	CLIENT_MODE_CTX = 1
	CLIENT_MODE_SET = 2
)

type RedisDistributedJobWorker struct {
	base.BaseStarter
	ticker    *time.Ticker
	mutex     *redsync.Mutex
	mode      int
	optPrefix string
	client    redis.UniversalClient
	taskKey   string
	taskFunc  func()
}

func NewRedisDistributedJobWorker(taskKey string, taskFunc func(), mode int) *RedisDistributedJobWorker {
	return &RedisDistributedJobWorker{
		taskFunc: taskFunc,
		taskKey:  taskKey,
		mode:     mode,
	}
}

func (s *RedisDistributedJobWorker) Init(ctx base.StarterContext) {
	var client redis.UniversalClient
	if s.mode == CLIENT_MODE_NEW {
		opts := &redis.Options{
			//Addr:     conf.GetDefault("redis.addr", "127.0.0.1:6379"),
			//Password: conf.GetDefault("redis.password", ""),
			//DB:       conf.GetIntDefault("redis.database", 0),
			//PoolSize: conf.GetIntDefault("redis.poolSize", 3),
		}
		prefix := s.optPrefix
		if prefix == "" {
			prefix = "redis"
		}
		err := kvs.Unmarshal(ctx.Props(), opts, prefix)
		if err != nil {
			panic(err)
		}
		log.Infof("connecting redis: %+v", opts)
		client = redis.NewClient(opts)
		log.Info("connected redis:", client)
	}
	if s.mode == CLIENT_MODE_CTX {
		client = Rdb()
	}
	if s.mode == CLIENT_MODE_SET {
		client = s.client
	}

	pool := goredis.NewPool(client)
	rsync := redsync.New(pool)
	ip, err := utils.GetExternalIP()
	if err != nil {
		ip = "127.0.0.1"
	}

	s.mutex = rsync.NewMutex("lock:"+s.taskKey,
		redsync.WithExpiry(50*time.Second),
		redsync.WithTries(10),
		redsync.WithGenValueFunc(func() (s string, e error) {
			now := time.Now()
			log.Infof("节点%s正在执行该任务", ip)
			return fmt.Sprintf("%d:%s", now.Unix(), ip), nil
		}))
}

func (s *RedisDistributedJobWorker) Start(ctx base.StarterContext) {
	interval := ctx.Props().GetDurationDefault(fmt.Sprintf("job.%s.interval", s.taskKey), time.Minute)

	s.ticker = time.NewTicker(interval)

	go func() {
		for {
			c := <-s.ticker.C
			do := func() {
				err := s.mutex.Lock()
				s.mutex.Unlock()
				defer s.mutex.Unlock()
				if err == nil {
					log.Info("task starting..., ", c)
					s.taskFunc()
				} else {
					log.Info("已经有节点在处理该任务.")
				}
			}
			do()
		}
	}()
}
func (s *RedisDistributedJobWorker) Stop(ctx base.StarterContext) {
	s.ticker.Stop()
}
